import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

import java.io.Serializable;
import java.time.Duration;

public class Xwm {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


//        DataStreamSource<String> stream = env.socketTextStream("doitedu", 9999);
//        SingleOutputStreamOperator<Some> map = stream.map(s -> JSON.parseObject(s, Some.class));
//
//        SingleOutputStreamOperator<Some> wm = map.assignTimestampsAndWatermarks(WatermarkStrategy.<Some>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Some>() {
//            @Override
//            public long extractTimestamp(Some some, long l) {
//                return some.ts;
//            }
//        }));



        // 构建一个kafkaSource对象
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setGroupId("g001")
                .setClientIdPrefix("flink-c-")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setBootstrapServers("doitedu:9092")
                .setTopics("odd")
                .build();


        // 用env使用该source获取流
        DataStreamSource<String> wm = env.fromSource(source, WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
            @Override
            public long extractTimestamp(String some, long l) {

                long ts = JSON.parseObject(some).getLongValue("ts");

                return ts;
            }
        }), "随便");


        SingleOutputStreamOperator<Some2> map = wm.map(s -> JSON.parseObject(s, Some2.class));


        tenv.createTemporaryView("wm",map, Schema.newBuilder()
                        .column("id", DataTypes.INT())
                        .column("name", DataTypes.STRING())
                        .columnByMetadata("rt",DataTypes.TIMESTAMP(3),"rowtime")
                        //.watermark("rt","source_watermark()")
                        .watermark("rt","rt")
//                        .columnByExpression("rt","to_timestamp_ltz(ts,3)")
//                        .watermark("rt","rt")
                .build());

        //tenv.executeSql("select * from wm").print();

        tenv.executeSql("select window_start,window_end,name,count(1) as cnt " +
                "from table(tumble(table wm,descriptor(rt),interval '5' second)) " +
                "group by window_start,window_end,name").print();



    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Some implements Serializable{
        private int id;
        private String name;
        private long ts;
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Some2 implements Serializable{
        private int id;
        private String name;
    }
}
